package com.atguigu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.CartAddUuBean;
import com.atguigu.util.DateFormatUtil;
import com.atguigu.util.KafkaUtil;
import com.atguigu.util.MyClickHouseUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author hfh
 * @version 1.0
 * @date 2022/8/25 15:29
 */
public class DwsUserCartAddUuWindow {
    public static void main(String[] args) throws Exception {
        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置状态后端
        /*env.enableCheckpointing(5*60*1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setCheckpointTimeout(3*60*1000L);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

        //TODO 1 从kafka加购明细主题读取数据
        String topicName = "dwd_user_cart_add";
        String groupId = "dws_trade_cart_add_uu_window";
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));
//        streamSource.print();

        //TODO 2 转换数据结构  将流数据转为json
        SingleOutputStreamOperator<JSONObject> map = streamSource.map(JSONObject::parseObject);

        //TODO 3 设置水位线
        SingleOutputStreamOperator<JSONObject> timestampsAndWatermarks = map.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3L))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts") * 1000L;
                    }
                }));

        //TODO 4 按照用户uid分组
        KeyedStream<JSONObject, String> keyedStream = timestampsAndWatermarks.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                JSONObject data = value.getJSONObject("data");
                String userId = data.getString("user_id");
                return userId;
            }
        });

        //TODO 5 过滤独立用户加购记录
        SingleOutputStreamOperator<CartAddUuBean> flatMap = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, CartAddUuBean>() {
            private ValueState<String> valueState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> descriptor = new ValueStateDescriptor<>("lastState", String.class);
                descriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1L))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                        .build());
                valueState = getRuntimeContext().getState(descriptor);
            }

            @Override
            public void flatMap(JSONObject value, Collector<CartAddUuBean> out) throws Exception {
                String lastState = valueState.value();
                long ts = value.getLong("ts") * 1000L;
                String lastDt = DateFormatUtil.toDate(value.getLong("ts") * 1000L);
                Long cartAddUuCt = 0L;
                if (lastState == null || !lastState.equals(lastDt)) {
                    valueState.update(lastDt);
                    cartAddUuCt = 1L;
                }
                if (cartAddUuCt == 1) {
                    out.collect(new CartAddUuBean("", "", cartAddUuCt, ts));
                }
            }
        });

        //TODO 6 开窗 聚合
        AllWindowedStream<CartAddUuBean, TimeWindow> windowedStream = flatMap.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L))).allowedLateness(org.apache.flink.streaming.api.windowing.time.Time.seconds(2L));

        SingleOutputStreamOperator<CartAddUuBean> reduce = windowedStream.reduce(new ReduceFunction<CartAddUuBean>() {
            @Override
            public CartAddUuBean reduce(CartAddUuBean value1, CartAddUuBean value2) throws Exception {
                value1.setCartAddUuCt(value1.getCartAddUuCt() + value2.getCartAddUuCt());
                return value1;
            }
        }, new AllWindowFunction<CartAddUuBean, CartAddUuBean, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<CartAddUuBean> values, Collector<CartAddUuBean> out) throws Exception {
                CartAddUuBean next = values.iterator().next();
                next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                out.collect(next);
            }
        });

        reduce.print("reduce>>>>>>>");

        //TODO 7 写出到clickHouse
        reduce.addSink(MyClickHouseUtil.<CartAddUuBean>getSink("insert into dws_trade_cart_add_uu_window values (?,?,?,?)"));

        env.execute(groupId);
    }
}
